package com.share.pulsar.rest;

import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author caifan
 * @created 2021/5/19
 * description:
 */
@RestController
@RequestMapping("/consumer")
public class PulsarConsumerRest {
    @Autowired
    private PulsarClient pulsarClient;

    @GetMapping("/simple")
    public String simpleConsumer() throws PulsarClientException {
        Consumer<String> subscribe = pulsarClient.newConsumer(Schema.STRING)
                .topic("simple-topic")
                .consumerName("simple-consumer")
                .subscriptionName("simple-subscription")
                .subscribe();
        Message receive = subscribe.receive();
        subscribe.acknowledgeAsync(receive);
        return new String(receive.getData());
    }
}
